package com.qf.cx.simplenms.link.impl;

import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import javax.annotation.PostConstruct;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.PropertySource;
import org.springframework.core.env.Environment;
import org.springframework.stereotype.Component;

import com.qf.cx.simplenms.link.api.LinkManager;
import com.qf.cx.simplenms.link.rai.RaiLibrary;
import com.qf.cx.simplenms.link.utils.Utils;
import com.sun.jna.ptr.IntByReference;

@Component
@PropertySource("classpath:link.properties")
class PhysicalLinkMaintainerImpl implements PhysicalLinkMaintainerIF {
	private final Logger LOG = LoggerFactory.getLogger(this.getClass());
	@Autowired
	private RaiLibrary libInstance;
	@Autowired
	private Environment env;
	private int keepAlive;
	private int connectTimeout;
	private Map<String, Integer> listenMap = new ConcurrentHashMap<>();
	private Map<String, PhysicalLinkIF> successfulLinks = new ConcurrentHashMap<>();
	private Set<LinkManager.ReaderSource> allReaderSources = new CopyOnWriteArraySet<>();
	private final ExecutorService acceptListenPool = Executors.newCachedThreadPool(new ThreadFactory() {
		@Override
		public Thread newThread(Runnable r) {
			return new Thread(r, "link-accept task");
		}
	});

	public RaiLibrary getLibInstance() {
		return libInstance;
	}

	public void setLibInstance(RaiLibrary libInstance) {
		this.libInstance = libInstance;
	}

	@PostConstruct
	public void init() {
		keepAlive = Integer.parseInt(env.getProperty("link_keepalive"));
		connectTimeout = Integer.parseInt(env.getProperty("connect_timeout"));
		LOG.info("client connect task start !");
		new Thread(new ConnectTask(), "link-connect task").start();
		LOG.info("disconnect task start !");
		new Thread(new DisconnectTask(), "link-disconnect task").start();
	}

	@Override
	public PhysicalLinkIF getLink(LinkManager.ReaderSource rs) {

		addToAllReaderSources(rs);
		final PhysicalLinkIF physicalLinkIF = successfulLinks.get(rs.getReaderIdentifier());
		if (null == physicalLinkIF) {
			return NullPhysicalLinkImpl.INSTANCE;
		}
		if (physicalLinkIF.isOk()) {
			return physicalLinkIF;
		}
		LOG.info("ReaderSource:{} phy link is not ok,disconnect it!", rs);
		physicalLinkIF.disconnect();
		successfulLinks.remove(rs.getReaderIdentifier());
		return NullPhysicalLinkImpl.INSTANCE;
	}

	private void addToAllReaderSources(LinkManager.ReaderSource rs) {
		allReaderSources.add(rs);
	}

	@Override
	public void removeFromAllReaderSources(LinkManager.ReaderSource rs) {
		if (allReaderSources.contains(rs)) {
			LOG.info("ReaderSource:{} remove for all reader sources!", rs);
			allReaderSources.remove(rs);
		}
	}

	@Override
	public void listen(String ip, int port) {
		IntByReference listenFd = new IntByReference();
		final String url = toListenUrl(ip, port);
		final int listenRtn = libInstance.rai_startListen(url, listenFd);
		if (0 == listenRtn) {
			listenMap.put(url, listenFd.getValue());
			acceptListenPool.execute(new AcceptConnectionTask(url));
			LOG.info("start listen url : {}", url);
			return;
		}
		throw new RuntimeException("监听端口失败ip:" + ip + " port is:" + port);
	}

	private String toListenUrl(String ip, int port) {
		return "net:listen://port:" + port + toKeepAlive(keepAlive);
	}
	
	private String toKeepAlive(int keepAlive) {
		if (keepAlive == 0) {
			return "";
		} else {
			return ";keepalive:" + keepAlive;
		}
	}

	class AcceptConnectionTask implements Runnable {
		String listenUrl;

		public AcceptConnectionTask(String listenUrl) {
			this.listenUrl = listenUrl;
		}

		@Override
		public void run() {
			while (true) {
				Utils.sleepms(10);
				IntByReference connectFd = new IntByReference();
				final int getListenLinkResult = libInstance.rai_getListenLink(listenMap.get(listenUrl), connectFd);
				if (0 != getListenLinkResult) {
					continue;
				} else {
					final String deviceSn = raiGetdeviceSN(connectFd.getValue());
					if (null != deviceSn) {
						LOG.info("accept a new phy link,SN = {},handler = {}", deviceSn, connectFd.getValue());
						PhysicalLinkIF phyLink = successfulLinks.get(deviceSn);

						if (phyLink != null) {

							if (phyLink.linkstatus() != 0) {
								phyLink.disconnect();
								LOG.info("a old phy link SN = {} handle status not ok,disconnect old link", deviceSn);
								successfulLinks.put(deviceSn, new PhysicalLinkImpl(connectFd.getValue(), libInstance));
							} else {
								new PhysicalLinkImpl(connectFd.getValue(), libInstance).disconnect();
								LOG.info("a new phy link SN = {} handle has exsit,disconnect the link", deviceSn);
							}
						} else {
							successfulLinks.put(deviceSn, new PhysicalLinkImpl(connectFd.getValue(), libInstance));
						}

					}
				}
			}
		}
	}

	private String raiGetdeviceSN(int fd) {
		String deviceSN;
		byte[] pSN = new byte[8];
		int ret = libInstance.rai_getDeviceSN(fd, pSN);
		if (ret == 0) {
			deviceSN = Utils.arr2String(Utils.byte2uchar(pSN));
			LOG.debug("device sn:{}.", deviceSN);
			return deviceSN;
		} else {
			LOG.error("get deviceSN error : " + ret);
			return null;
		}
	}

	class ConnectTask implements Runnable {
		@Override
		public void run() {
			while (true) {
				Utils.sleepms(5000);
				allReaderSources.forEach(rs -> {
					if (!isNeedConnect(rs)) {
						return;
					}
					IntByReference fd = new IntByReference();
					final String url = toUrl(rs);
					LOG.info("phy link connecting ,url = {},handler = {}", url, fd.getValue());
					final int retOfConnect = libInstance.rai_connect(url, fd);
					if (retOfConnect == 0) {
						LOG.info("a new phy link connect,url = {},handler = {}", url, fd.getValue());
						successfulLinks.put(rs.getReaderIdentifier(), new PhysicalLinkImpl(fd.getValue(), libInstance));
					} else {
						LOG.warn("phy link connect fail,url = {}", url);
					}
				});
			}
		}

		private boolean isNeedConnect(LinkManager.ReaderSource rs) {
			if (successfulLinks.containsKey(rs.getReaderIdentifier())) {
				return false;
			}
			if (rs.getLinkType() == LinkManager.LinkType.CLIENT) {
				return false;
			}
			return true;
		}

		private String toUrl(LinkManager.ReaderSource rs) {
			return "net:connect://ip:" + rs.getIp() + ";" + "port:" + rs.getPort() + ";"
					+ "timeout:" + connectTimeout + toKeepAlive(keepAlive);
		}
	}

	class DisconnectTask implements Runnable {
		@Override
		public void run() {
			while (true) {
				Utils.sleepms(5000);
				for (Entry<String, PhysicalLinkIF> e : successfulLinks.entrySet()) {
					if (!isLinkInReadrSources(e)) {
						successfulLinks.entrySet().remove(e);
						e.getValue().disconnect();
					}
				}
			}
		}

		private boolean isLinkInReadrSources(Entry<String, PhysicalLinkIF> e) {
			boolean found = false;
			for (LinkManager.ReaderSource r : allReaderSources) {
				if (r.getReaderIdentifier().equals(e.getKey())) {
					found = true;
					break;
				}
			}
			return found;
		}
	}
}